#ifndef __M__CONNECTION_H_
#define __M__CONNECTION_H_

#include "channel.hpp"
#include "worker.hpp"

#include "Logging.h"
#include "Mutex.h"
#include "TcpClient.h"
#include "EventLoop.h"
#include "TcpConnection.h"
#include "CountDownLatch.h"
#include "EventLoopThread.h"
#include "../third/muduo/proto/dispatcher.h"

using namespace mq_message;
using namespace mq;
using namespace std;
using namespace muduo::net;

namespace mq
{
    // 客户端连接类：封装客户端网络通信连接，对上只提供被连接管理的信道的概念
    // 弱化了客户端的概念，所有服务通过信道提供
    class Connection
    {
    public:
        using ptr = shared_ptr<Connection>;
        using MessagePtr = shared_ptr<google::protobuf::Message>;
        using basicCommonResponsePtr = shared_ptr<basicCommonResponse>;
        using basicConsumeResponsePtr = shared_ptr<basicConsumeResponse>;

        Connection(const string &svr_ip, int svr_port,const AsyncWorker::ptr& worker)
            : _latch(1),
              _client(worker->loopthread.startLoop(), muduo::net::InetAddress(svr_ip, svr_port), "Connection"),
              _dispatcher(std::bind(&Connection::onUnknownMessage, this, placeholders::_1, placeholders::_2, placeholders::_3)),
              _codec(make_shared<ProtobufCodec>(std::bind(&ProtobufDispatcher::onProtobufMessage, &_dispatcher, placeholders::_1, placeholders::_2, placeholders::_3))),
              _worker(worker),
              _channel_manager(make_shared<ChannelManager>())
        {
            // 注册响应处理回调
            _dispatcher.registerMessageCallback<basicCommonResponse>(std::bind(&Connection::basicResponse, this,
                                                                               placeholders::_1, placeholders::_2, placeholders::_3));
            _dispatcher.registerMessageCallback<basicConsumeResponse>(std::bind(&Connection::consumeResponse, this,
                                                                               placeholders::_1, placeholders::_2, placeholders::_3));

            // 参数绑定回调
            _client.setConnectionCallback(std::bind(&Connection::onConnection, this, std::placeholders::_1));
            _client.setMessageCallback(std::bind(&ProtobufCodec::onMessage, _codec, placeholders::_1, placeholders::_2, placeholders::_3));
        }
	~Connection()
	{
		_client.disconnect();
	}
        void connect()
        {
            _client.connect(); // 连接服务器
            _latch.wait();     // 主线程进入阻塞，等待连接成功后，调用onConnection回调唤醒
        }
        Channel::ptr openChannel() // 打开信道
        {
            // 1.客户端创建信道
            Channel::ptr channel = _channel_manager->openChanel(_server_conn, _codec);
            // 2.服务端创建对应信道
            if (!channel->openChannel())
            {
                DLOG("打开信道失败");
                _channel_manager->closeChannel(channel);
                return Channel::ptr();
            }
            return channel;
        }
        void closeChannel(const Channel::ptr &channel) // 关闭信道
        {
            // 1.服务端关闭对应信道
            channel->closeChannel();
            // 2.客户端关闭信道
            _channel_manager->closeChannel(channel);
        }

    private:
        void onUnknownMessage(const muduo::net::TcpConnectionPtr &conn, const MessagePtr &message, muduo::Timestamp)
        { // 收到未定义的消息类型时，回调处理
            cout << "onUnknownMessage:" << message->GetTypeName() << endl;
        }
        void onConnection(const muduo::net::TcpConnectionPtr &conn)
        { // 成功连接服务器时回调
            if (conn->connected())
            {
                _server_conn = conn;
                _latch.countDown(); // 唤醒主线程的阻塞
            }
            else
            {
                // 连接关闭
                _server_conn.reset(); // 重置连接
            }
        }
        void basicResponse(const TcpConnectionPtr &conn, const basicCommonResponsePtr &resp, muduo::Timestamp) // 基础响应处理
        {
            // 1.找到对应信道
            string cid = resp->cid();
            Channel::ptr channel = _channel_manager->getChannel(cid);
            if (channel.get() == nullptr)
            {
                DLOG("未找到对应信道: %s", cid);
                return;
            }
            // 2.将响应放入响应列表，并唤醒对应信道
            channel->putBasicCommonResponse(resp);
        }
        void consumeResponse(const TcpConnectionPtr &conn, const basicConsumeResponsePtr &resp, muduo::Timestamp) // 消息推送响应处理
        {
            // 1.找到对应信道
            string cid = resp->cid();
            Channel::ptr channel = _channel_manager->getChannel(cid);
            if (channel.get() == nullptr)
            {
                DLOG("未找到对应信道: %s", cid);
                return;
            }
            // 2.封装异步任务(消息处理)，放入线程池
            _worker->tpool.PushTask([channel, resp]()
                                    { channel->consume(resp); });
        }

    private:
        muduo::CountDownLatch _latch;              // 防止连接还没建立成功，就使用未定义的连接对象
        muduo::net::TcpClient _client;             // 客户端类
        muduo::net::TcpConnectionPtr _server_conn; // 服务器连接对象

        ProtobufDispatcher _dispatcher; // 请求分发器：注册请求处理函数
        ProtobufCodecPtr _codec;        // protobuf协议处理器：针对数据协议处理

        AsyncWorker::ptr _worker;             // 异步工作线程对象
        ChannelManager::ptr _channel_manager; // 信道管理句柄
    };
}

#endif
